package pku;

import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 生产者
 */
public class Producer {

	Map<String, FileOutputStream> foses = new HashMap<>();
	Map<String, FileChannel> channels = new HashMap<>();
	Map<String, ByteBuffer> buffers = new HashMap<>();

	String path = "";

	static AtomicInteger num = new AtomicInteger();

	String produceName = null;

	public Producer() {
		num.incrementAndGet();
		produceName = "produce" + num.get();

	}

	// 生成一个指定topic的message返回
	public ByteMessage createBytesMessageToTopic(String topic, byte[] body) {
		File fileTopic = new File(Constant.PATH + topic);
		if (!fileTopic.exists()) {
			fileTopic.mkdir();
		}
		path = Constant.PATH + topic + "/";
		String filename = topic + "_" + produceName;
		if (!foses.containsKey(filename)) {
			try {
				String filePath = path + filename + ".txt";
				File file = new File(filePath);
				file.createNewFile();
				FileOutputStream fos = new FileOutputStream(file, true);
				FileChannel channel = fos.getChannel();
				ByteBuffer buffer = ByteBuffer.allocate(1024 * 512);
				foses.put(filename, fos);
				channels.put(filename, channel);
				buffers.put(filename, buffer);
			} catch (FileNotFoundException e) {
				e.printStackTrace();
			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}

		ByteMessage msg = new DefaultMessage(body);
		msg.putHeaders(MessageHeader.TOPIC, topic);
		return msg;
	}

	// 将message发送出去
	public void send(ByteMessage defaultMessage) {
		String topic = defaultMessage.headers().getString(MessageHeader.TOPIC);
		String filename = topic + "_" + produceName;
		ByteBuffer buffer = buffers.get(filename);
		if (buffer.position() > 1024 * 256) {
			buffer.flip();
			try {
				channels.get(filename).write(buffer);
			} catch (IOException e) {
				e.printStackTrace();
			}
			buffer.clear();
			
			
			short mark = 0;
			KeyValue headers = defaultMessage.headers();
			byte[] body = defaultMessage.getBody();

			for (int i = 0; i < 16; i++) {
				String key = MessageHeaderConstant.getHeaderInverseMapValue(i);
				Object object = headers.getObj(key);
				if (object != null) {
					mark = MyUtilTool.setBit(mark, i);
				}
				
			}
			buffer.putShort(mark);
			for (int i = 0; i < 16; i++) {
				String key = MessageHeaderConstant.getHeaderInverseMapValue(i);
				Object object = headers.getObj(key);

				MyUtilTool.setBit(mark, MessageHeaderConstant.getHeaderMapValue(key));
				if (object != null) {
					if (object instanceof Integer) {
						int value = (int) object;
						buffer.putInt(value);
					} else if (object instanceof Double) {
						double value = (double) object;
						buffer.putDouble(value);
					} else if (object instanceof Long) {
						long value = (long) object;
						buffer.putLong(value);
					} else if (object instanceof String) {
						String value = (String) object;
						byte[] valueBytes = value.getBytes();
						buffer.putInt(valueBytes.length);
						buffer.put(valueBytes);
					}
				}
			}
			buffer.putInt(body.length);
			buffer.put(body);
		} else {
			short mark = 0;
			KeyValue headers = defaultMessage.headers();
			byte[] body = defaultMessage.getBody();

			for (int i = 0; i < 16; i++) {
				String key = MessageHeaderConstant.getHeaderInverseMapValue(i);
				Object object = headers.getObj(key);
				if (object != null) {
					mark = MyUtilTool.setBit(mark, i);
				}
				
			}
			buffer.putShort(mark);
			for (int i = 0; i < 16; i++) {
				String key = MessageHeaderConstant.getHeaderInverseMapValue(i);
				Object object = headers.getObj(key);

				MyUtilTool.setBit(mark, MessageHeaderConstant.getHeaderMapValue(key));
				if (object != null) {
					if (object instanceof Integer) {
						int value = (int) object;
						buffer.putInt(value);
					} else if (object instanceof Double) {
						double value = (double) object;
						buffer.putDouble(value);
					} else if (object instanceof Long) {
						long value = (long) object;
						buffer.putLong(value);
					} else if (object instanceof String) {
						String value = (String) object;
						byte[] valueBytes = value.getBytes();
						buffer.putInt(valueBytes.length);
						buffer.put(valueBytes);
					}
				}
			}
			buffer.putInt(body.length);
			buffer.put(body);
		}
	}

	// 处理将缓存区的剩余部分
	public void flush() throws Exception {
		for (Map.Entry<String, ByteBuffer> entry : buffers.entrySet()) {
			entry.getValue().flip();
			channels.get(entry.getKey()).write(entry.getValue());
			entry.getValue().clear();
			channels.get(entry.getKey()).close();
			foses.get(entry.getKey()).close();
		}
	}
}
